package com.dongnao.network.nio.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 工作线程池下的多Reactor模式
 */
public class WorkerThreadPoolMultipleReactor {

    static ExecutorService mainThreadPool = CustomThreadPool.buildPool();

    public static void main(String[] args) {
        WorkerThreadPoolMultipleReactor reactor = new WorkerThreadPoolMultipleReactor();
        reactor.init(8080);
    }

    final int subReaLen = 8;
    SubReactor[] subReaGroup = new SubReactor[subReaLen];

    private void init(int port) {
        try {
            for (int i = 0; i < subReaGroup.length; i++) {
                subReaGroup[i] = new SubReactor();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            MainReactor main = new MainReactor();
            ServerSocketChannel serverSocket = ServerSocketChannel.open();
            serverSocket.bind(new InetSocketAddress(port));
            System.out.println("服务器启动，端口" + port);
            main.register(serverSocket);
            mainThreadPool.submit(main);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 处理多个客户端并发请求
     * MainReactor只处理Accept IO事件
     */
    private class MainReactor implements Runnable {
        final Selector sel;
        ServerSocketChannel serverSocket;

        public MainReactor() throws IOException {
            sel = Selector.open();
        }

        public void register(ServerSocketChannel serverSocketChannel) {
            try {
                serverSocket = serverSocketChannel;
                serverSocket.configureBlocking(false);
                //获取serverSocket在Selector中注册的标识sk，sk绑定了serverSocket与Selector的关系，维护channel事件
                SelectionKey sk = serverSocket.register(sel, SelectionKey.OP_ACCEPT);
                //附加一个Runnable的Accept事件处理者
                sk.attach(new Acceptor(sel, serverSocket));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    sel.select();
                    Set<SelectionKey> selKeySet = sel.selectedKeys();
                    Iterator<SelectionKey> it = selKeySet.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        System.out.println("key isAcceptable: " + key.isAcceptable()
                                + ", isConnectable: " + key.isConnectable()
                                + ", isReadable: " + key.isReadable()
                                + ", isWritable: " + key.isWritable()
                                + ", isValid: " + key.isValid());
                        dispatch(key);
                    }
                    selKeySet.clear();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private void dispatch(SelectionKey selectionKey) {
            Runnable r = (Runnable) selectionKey.attachment();
            if (r != null) {
                r.run();
            }
        }
    }

    /**
     * 处理多个客户端并发请求
     */
    private class SubReactor implements Runnable {
        final Selector sel;
        SocketChannel socket;

        public SubReactor() throws IOException {
            sel = Selector.open();
        }

        public void register(SocketChannel socketChannel) {
            socket = socketChannel;
            try {
                socket.configureBlocking(false);
                SelectionKey key = socket.register(sel, SelectionKey.OP_READ);
                new Handler(key);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    sel.select();
                    Set<SelectionKey> selKeySet = sel.selectedKeys();
                    Iterator<SelectionKey> it = selKeySet.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        System.out.println("key isAcceptable: " + key.isAcceptable()
                                + ", isConnectable: " + key.isConnectable()
                                + ", isReadable: " + key.isReadable()
                                + ", isWritable: " + key.isWritable()
                                + ", isValid: " + key.isValid());
                        dispatch(key);
                    }
                    selKeySet.clear();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private void dispatch(SelectionKey selectionKey) {
            Runnable r = (Runnable) selectionKey.attachment();
            if (r != null) {
                r.run();
            }
        }
    }

    private class Acceptor implements Runnable {
        final Selector sel;
        final ServerSocketChannel serverSocket;
        final SubReactor[] subRea;
        final AtomicInteger seq = new AtomicInteger();

        public Acceptor(Selector selector, ServerSocketChannel serverSocketChannel) {
            sel = selector;
            serverSocket = serverSocketChannel;
            subRea = subReaGroup;
        }

        /**
         * 由Reactor的dispatch调用，处理客户端的连接请求accept事件
         */
        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocket.accept();
                System.out.println("----Acceptor#run()----channel: " + (channel == null ? "null" : channel));
                if (channel != null) {
                    System.out.println("收到新的客户端连接：" + channel);
                    //new Handler(sel, channel);

                    //不再直接绑定Handler，而是绑定到某个SubReactor上去
                    int index = seq.getAndIncrement();
                    int idx = index % subRea.length;
                    subRea[idx].register(channel);
                    mainThreadPool.submit(subRea[idx]);

                    //避免溢出
                    if (seq.get() == Integer.MAX_VALUE >> 1) {
                        seq.set(0);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 客户端读写事件处理器
     */
    private class Handler implements Runnable {
        final SocketChannel socket;
        final SelectionKey sk;
        final int MAXIN = 1024;
        ByteBuffer input = ByteBuffer.allocate(MAXIN);
        ExecutorService threadPool = CustomThreadPool.buildPool();

        public Handler(SelectionKey key) throws IOException {
            System.out.println("----Handler()----");
            socket = (SocketChannel) key.channel();
            sk = key;
            sk.attach(this);//绑定自己为读写事件的处理器
            //selector.wakeup();//激活在select()方法上的线程，告诉它感兴趣的事件发生变化了
        }

        @Override
        public void run() {
            System.out.println("----Handler#run()----");
            try {
                if (sk.isReadable()) {
                    read();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        void read() throws IOException {
            int len = socket.read(input);
            if (inputIsComplete()) {
                input.flip();
                //process();//读完数据，进行业务逻辑处理
                threadPool.execute(new Processer(sk, input));
                input.clear();
            }
            if (len < 0) {
                sk.cancel();
            }
        }

        private boolean inputIsComplete() {
            return input.position() > 0;
        }
    }

    private class Processer implements Runnable {
        final ByteBuffer input;
        final SelectionKey sk;

        public Processer(SelectionKey selKey, ByteBuffer in) {
            input = ByteBuffer.wrap(in.array(), 0, in.limit());
            sk = selKey;
        }

        @Override
        public void run() {
            //模拟业务逻辑处理
            System.out.println("收到数据：" + new String(input.array(), input.position(), input.limit()));
//            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(300));
            String msg = "Now time is : " + (new Date()).toString();
            ByteBuffer output = ByteBuffer.allocate(1024);
            output.put(msg.getBytes());
            try {
                output.flip();
//                new Sender(sk, output).run();
                new Sender(sk, output);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private class Sender implements Runnable {
        final SocketChannel socket;
        ByteBuffer output;
        SelectionKey sk;
        final Object oldAttachment;

        public Sender(SelectionKey selKey, ByteBuffer respMsg) {
            socket = (SocketChannel) selKey.channel();
            output = ByteBuffer.wrap(respMsg.array(), 0, respMsg.limit());
            sk = selKey;
            sk.interestOps(sk.interestOps() | SelectionKey.OP_WRITE);
            oldAttachment = sk.attach(this);
            sk.selector().wakeup();
        }

        @Override
        public void run() {
            try {
                if (sk.isWritable()) {
                    socket.write(output);
                    if (outputIsComplete()) {
                        output.clear();
                        sk.attach(oldAttachment);
                        //只有写事件操作完成了，才取消
                        sk.interestOps(sk.interestOps() & ~SelectionKey.OP_WRITE);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private boolean outputIsComplete() {
            return !output.hasRemaining();
        }
    }
}
